﻿// Copyright (c) .NET Core Community. All rights reserved.
// Licensed under the MIT License. See License.txt in the project root for license information.

using System;
using System.Collections.Generic;
using System.Linq;
using System.Reflection;
using System.Text;
using System.Threading;
using System.Xml.Linq;
using MQTTnet;
using MQTTnet.Client;
#if !NET_FULL
using MQTTnet.Client.Options;
using MQTTnet.Client.Disconnecting;
using MQTTnet.Client.Connecting;
using MQTTnet.Client.Receiving;
using MQTTnet.Formatter;
#else
using MQTTnet.Serializer;
#endif
using DotNetFramework.CAP.Abstractions;
using DotNetFramework.CAP.Internal;
using DotNetFramework.CAP.Models;
using MQTTnet.Serializer;
using DotNetFramework.CAP.Messages;
using DotNetFramework.CAP.Mqtt;
using Newtonsoft.Json;
using DotNetFramework.CAP.NodeDiscovery;
using System.Net.Sockets;
using System.Security.Cryptography.X509Certificates;
using DotNetFramework.CAP.Core.Logger;
//using DotNetFramework.CAP.Messages;
//using DotNetFramework.CAP.Transport;
//using Microsoft.Extensions.Options;

namespace DotNetFramework.CAP.Mqtt
{
    public class MqttConsumerClient : IConsumerClient
    {
        private static readonly SemaphoreSlim ConnectionLock = new SemaphoreSlim(initialCount: 1, maxCount: 1);


        private string _groupId;
        private readonly MqttOptions _mqttOptions;
        private IMqttClient _mqttClient;
        IContentSerializer _serializer;
        ILogger _logger;

        public string ServersAddress { get; private set; }

        public string ClientId ;
        public MqttConsumerClient(ILogger logger, string groupId, MqttOptions options, IContentSerializer serializer)
        {
            ClientId = Guid.NewGuid().ToString();
            _logger = logger;
            _groupId = groupId;
            _mqttOptions = options;
            ServersAddress = options.Server;
            _serializer = serializer;
        }

        public event EventHandler<MessageContext> OnMessageReceived;

        public event EventHandler<LogMessageEventArgs>? OnLog;

        //public BrokerAddress BrokerAddress => new("Kafka", _kafkaOptions.Servers);

        public ICollection<string> FetchTopics(IEnumerable<string> topicNames)
        {
            if (topicNames == null)
            {
                throw new ArgumentNullException(nameof(topicNames));
            }

            var regexTopicNames = topicNames.Select(Infrastructure.Helper.WildcardToRegex).ToList();

            return regexTopicNames;
        }

        IEnumerable<string> _topics;
        public void Subscribe(IEnumerable<string> topics)
        {
            if (topics == null)
            {
                throw new ArgumentNullException(nameof(topics));
            }
            _topics = topics;
            try
            {
                Connect();
            }
            catch (Exception e)
            {
                _logger.LogError(e, e.Message);
            }
        }
        bool isConnected
        {
            get
            {
                return _mqttClient != null && _isOpened; // _mqttClient.IsConnected;
            }
        }
        void OnConnectedSubscribe()
        {
            _logger.LogInformation($"{_groupId} c:{ClientId} connected.");
            foreach (string topic in _topics)
            {
                this._mqttClient?.SubscribeAsync(topic).Wait();
            }
        }
        public void Listening(TimeSpan timeout, CancellationToken cancellationToken)
        {
            try
            {
                this.Connect();
            }
            catch (Exception e)
            {
                _logger.LogError(e, e.Message);
            }
            while (!_isStoped)
            {
                cancellationToken.ThrowIfCancellationRequested();
                cancellationToken.WaitHandle.WaitOne(timeout);

                if (!isConnected)
                {
                    try
                    {
                        this.Connect();
                    }
                    catch (Exception)
                    {
                    }
                }
            }
        }

        void AddEvents(IMqttClient mqttClient)
        {
            //if (!_isConnectedOnce && isConnected)
            //{
#if NET40
            mqttClient.ConnectedHandler = new ConnectedHandler(this);
            mqttClient.DisconnectedHandler = new DisConnectedHandler(this);
            mqttClient.ApplicationMessageReceivedHandler = new ApplicationMessageReceivedHandler(this);
#elif NET_FULL
            mqttClient.Connected += HubConnection_Connected;
            mqttClient.ApplicationMessageReceived += HubConnection_ApplicationMessageReceived;
            mqttClient.Disconnected += HubConnection_Disconnected;
#else
            mqttClient.UseConnectedHandler((e) =>
           {
               _connected.Set();
               OnConnected();
           });
            mqttClient.UseApplicationMessageReceivedHandler((e) =>
            {
                OnApplicationMessageReceived(e);
            });
            mqttClient.UseDisconnectedHandler((e) =>
            {
                OnDisConnected();
            });
#endif
        }
        void UnEvents()
        {
            if (_mqttClient != null)
            {
#if NET_FULL
                _mqttClient.Connected -= HubConnection_Connected;
                _mqttClient.ApplicationMessageReceived -= HubConnection_ApplicationMessageReceived;
                _mqttClient.Disconnected -= HubConnection_Disconnected;
#endif

            }
        }
        volatile bool _isStoped = false;
        public void Dispose()
        {
            _isStoped = true;
            if (this._mqttClient == null)
            {
                return;
            }
            this._mqttClient.Dispose();
        }

        public bool Connect()
        {
            if (this._mqttClient != null && this._mqttClient.IsConnected)
            {
                return false;
            }
            ConnectionLock.Wait();
            try
            {
                var options = GetClientOption($"{_groupId}_{this.ClientId}", _mqttOptions.Server, _mqttOptions.Port, _mqttOptions.UserName, _mqttOptions.Password);
                if (this._mqttClient == null)
                {
                    this._mqttClient = new MqttFactory().CreateMqttClient();

                    AddEvents(_mqttClient);
                }
                this._mqttClient.ConnectAsync(options).GetAwaiter().GetResult();
                return true;
            }
            catch (Exception ex)
            {
                throw ex;
            }
            finally
            {
                ConnectionLock.Release();
            }
        }


        #region mqtt begin

#if NET40
        class ConnectedHandler : IMqttClientConnectedHandler
        {
            CommonMqttXPC _xpc;
            public ConnectedHandler(CommonMqttXPC mqttXPC)
            {
                _xpc = mqttXPC;
            }
            public Task HandleConnectedAsync(MqttClientConnectedEventArgs eventArgs)
            {
                _xpc.OnConnected();
                return Task.FromResult(0);
            }
        }
        class DisConnectedHandler: IMqttClientDisconnectedHandler
        {
            CommonMqttXPC _xpc;
            public DisConnectedHandler(CommonMqttXPC mqttXPC)
            {
                _xpc = mqttXPC;
            }

            public Task HandleDisconnectedAsync(MqttClientDisconnectedEventArgs eventArgs)
            {
                _xpc.OnDisConnected();
                return Task.FromResult(0);
            }
        }
        class ApplicationMessageReceivedHandler : IMqttApplicationMessageReceivedHandler
        {
            CommonMqttXPC _xpc;
            public ApplicationMessageReceivedHandler(CommonMqttXPC mqttXPC)
            {
                _xpc = mqttXPC;
            }

            public Task HandleApplicationMessageReceivedAsync(MqttApplicationMessageReceivedEventArgs eventArgs)
            {
                _xpc.OnApplicationMessageReceived(eventArgs);
                return Task.FromResult(0);
            }
        }
#else

        public IMqttClientOptions GetClientOption(string clientId, string ip, int port, string username, string password)
        {
            MqttClientOptionsBuilder oBuilder = null;
            if (_mqttOptions.Procotol == MqttProcotolType.tcp)
            {
                oBuilder = new MqttClientOptionsBuilder()
                    .WithTcpServer(ip, port)
                    .WithClientId(clientId)
                    .WithProtocolVersion(MqttProtocolVersion.V311)
                    .WithCommunicationTimeout(TimeSpan.FromSeconds(30));
            }
            else
            {
                oBuilder = new MqttClientOptionsBuilder()
                    .WithWebSocketServer($"ws://{ip}:{port}/{_mqttOptions.ProcotolPath}")
                    .WithClientId(clientId)
                    .WithProtocolVersion(MqttProtocolVersion.V311)
                    .WithCommunicationTimeout(TimeSpan.FromSeconds(30));
            }
            if (!string.IsNullOrEmpty(username))
                oBuilder.WithCredentials(username, password);
            var option = oBuilder.Build();
            SetBufferSize(ref option);
            return option;
        }

        public static int BUFFER_SIZE = 4 * 1024 * 1024;//4MB
        void SetBufferSize(ref IMqttClientOptions options)
        {
            if (options.ChannelOptions is MqttClientTcpOptions)
            {
                ((MqttClientTcpOptions)options.ChannelOptions).BufferSize = BUFFER_SIZE;
            }
        }
        public IMqttClientOptions GetClientOption(string clientId, string ip, int port,
        string name, string pwd, string pfxPath, string pfxPassword = "1234", bool allowUntrustedCertificates = true)
        {
            MqttClientOptionsBuilder oBuilder = null;

            List<X509Certificate> x509list = new List<X509Certificate>();
            x509list.Add(new X509Certificate(pfxPath, pfxPassword));

            if (_mqttOptions.Procotol == MqttProcotolType.tcp)
            {
                oBuilder = new MqttClientOptionsBuilder()
                   .WithProtocolVersion(MqttProtocolVersion.V311)//mqtt版本
                    .WithClientId(clientId)//随机id
                    .WithTcpServer(ip, port)//tls的ip和端口
                    .WithCommunicationTimeout(TimeSpan.FromSeconds(5))
                    .WithCredentials(name, pwd)//有证书
                    .WithProtocolVersion(MqttProtocolVersion.V311)
                    .WithTls(new MqttClientOptionsBuilderTlsParameters()
                    {
                        AllowUntrustedCertificates = allowUntrustedCertificates,//允许不受信任的证书
                        UseTls = true,//使用TLS
#if !NET_FULL
                        Certificates = x509list,//客户端证书
                        CertificateValidationHandler = delegate { return true; },//证书验证回调
                                                                                 //CertificateValidationCallback = cafileValidCallback,
#else
                        Certificates = (IEnumerable<IEnumerable<byte>>)x509list,//客户端证书
                        CertificateValidationCallback = delegate { return true; },//证书验证回调
                                                                                  //CertificateValidationCallback = cafileValidCallback,
#endif
                        IgnoreCertificateChainErrors = false,//忽略证书链错误
                        IgnoreCertificateRevocationErrors = false//忽略证书撤销错误
                    });
            }
            else
            {
                oBuilder = new MqttClientOptionsBuilder()
                    //.WithProtocolVersion(MqttProtocolVersion.V311)//mqtt版本
                    .WithClientId(clientId)//随机id
                .WithTcpServer(ip, port)//tls的ip和端口
                    .WithCredentials(name, pwd)//有证书
                    .WithWebSocketServer($"ws://{ip}:{port}/{_mqttOptions.ProcotolPath}")
                    .WithProtocolVersion(MqttProtocolVersion.V311)
                    .WithCommunicationTimeout(TimeSpan.FromSeconds(5))
                    .WithTls(new MqttClientOptionsBuilderTlsParameters()
                    {
                        AllowUntrustedCertificates = allowUntrustedCertificates,//允许不受信任的证书
                        UseTls = true,//使用TLS
#if !NET_FULL
                        Certificates = x509list,//客户端证书
                        CertificateValidationHandler = delegate { return true; },//证书验证回调
                                                                                 //CertificateValidationCallback = cafileValidCallback,
#else
                        Certificates = (IEnumerable<IEnumerable<byte>>)x509list,//客户端证书
                        CertificateValidationCallback = delegate { return true; },//证书验证回调
                                                                                  //CertificateValidationCallback = cafileValidCallback,
#endif
                        IgnoreCertificateChainErrors = false,//忽略证书链错误
                        IgnoreCertificateRevocationErrors = false//忽略证书撤销错误
                    });
            }

            var mqttOption = oBuilder.Build();
            SetBufferSize(ref mqttOption);
            return mqttOption;
        }
        private void HubConnection_ApplicationMessageReceived(object sender, MqttApplicationMessageReceivedEventArgs e)
        {
            OnApplicationMessageReceived(e);
        }
        private void HubConnection_Connected(object sender, MqttClientConnectedEventArgs e)
        {
            OnConnected();
        }
        private void HubConnection_Disconnected(object sender, MqttClientDisconnectedEventArgs e)
        {
            OnDisConnected();
        }
        volatile bool _isOpened=false;
        void OnDisConnected()
        {
            _logger.LogInformation($"{_groupId} c:{ClientId} disconnected.");
            _isOpened = false;
        }
        void OnConnected()
        {
            _isOpened = true ;
            OnConnectedSubscribe();
        }
        void OnApplicationMessageReceived(MqttApplicationMessageReceivedEventArgs args)
        {
            byte[] payload = args.ApplicationMessage.Payload;
            var json = Encoding.UTF8.GetString(payload);
            MqttMessage mqttMessage = JsonConvert.DeserializeObject<MqttMessage>(json);

            string msgName = args.ApplicationMessage.Topic;
            string msgGroup = _groupId;

            var dto = new CapMessageDto(Encoding.UTF8.GetString(mqttMessage.Body));
            //{[cap-msg-group, cap.queue.capweb251.v1.v1]}
            //{[cap-msg-name, sample.console.showtime]}
            var message = new MessageContext
            {
                Group = msgGroup,
                Name = msgName,
                Content = _serializer.Serialize(dto)
            };

            OnMessageReceived?.Invoke(args, message);
        }

        public void Commit()
        {
        }

        public void Reject()
        {
        }
#endif
        #endregion mqtt end
    }
}